package product;

import java.util.HashSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class NoMarketProvice {

	/**
	 * 统计没有农产品市场的省份 market 4,provice 5
	 * 
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		if (args.length != 2) {
			System.err.println("path err!");
			System.exit(-1);
		}
		Job job = new Job(new Configuration(), "nomarketProvice");
		job.setJarByClass(NoMarketProvice.class);

		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		job.setMapperClass(nomaps.class);
		job.setReducerClass(noreduce.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);

		job.waitForCompletion(true);
	}

	public static class nomaps extends Mapper<LongWritable, Text, Text, Text> {

		protected void map(
				LongWritable key,
				Text value,
				org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, Text, Text>.Context context)
				throws java.io.IOException, InterruptedException {
			String[] lines = value.toString().split("\t");
			String p = "test";
			if (lines.length == 1) {
				context.write(new Text(lines[0]), new Text(p));
			} else if (lines.length == 6) {
				context.write(new Text(lines[4]), new Text(lines[3]));
			}
		};
	}

	public static class noreduce extends
			Reducer<Text, Text, Text, NullWritable> {
		HashSet<String> set1 = new HashSet<String>();
		HashSet<String> set2 = new HashSet<String>();

		protected void reduce(
				Text k2,
				java.lang.Iterable<Text> values,
				org.apache.hadoop.mapreduce.Reducer<Text, Text, Text, NullWritable>.Context context)
				throws java.io.IOException, InterruptedException {
			boolean hasTest = false;
			set1.clear();
			for (Text text : values) {
				set1.add(text.toString());
				if(text.toString().equals("test")){
					hasTest = true;
				}
			}
			if (set1.size() < 2 && hasTest) {
				set2.add(k2.toString());
			}
		};
		 protected void
		 cleanup(org.apache.hadoop.mapreduce.Reducer<Text,Text,Text,NullWritable>.Context
		 context) throws java.io.IOException ,InterruptedException {
		 context.write(new Text(set2.toString()), NullWritable.get());
		 };
	}
}
